﻿using System;
using System.Collections;
using System.Threading;

namespace Aj.Core
{
    /// <summary>
    /// 管理线程池
    /// </summary>
    /// <remarks>lsj add 2011.11.07</remarks>
    public class ManagedThreadPool
    {
        #region [ 成员变量 ]

        /// <summary> 在线程池中需要支配的最大线程数.</summary>
        private const int _maxWorkerThreads = 10;

        /// <summary>所有等待去执行回调的列队.</summary>
        private static Queue _waitingCallbacks;

        /// <summary>
        /// Used to signal that a worker thread is needed for processing.  Note that multiple
        /// threads may be needed simultaneously and as such we use a semaphore instead of
        /// an auto reset event.
        /// </summary>
        private static Semaphore _workerThreadNeeded;

        /// <summary>线程池中需要支配的所有工作线程列表.</summary>
        private static ArrayList _workerThreads;

        /// <summary>当前活动的线程数目.</summary>
        private static int _inUseThreads;

        private static object _poolLock = new object();

        #endregion

        #region [ 构造函数 ]

        /// <summary>初始化线程池.</summary>
        static ManagedThreadPool()
        {
            Initialize();
        }

        /// <summary>初始化线程池.</summary>
        private static void Initialize()
        {
            _waitingCallbacks = new Queue();
            _workerThreads = new ArrayList();
            _inUseThreads = 0;

            // 创建我们的"thread needed"事件
            _workerThreadNeeded = new Semaphore(0);

            // 创建所有的工作线程
            for (int i = 0; i < _maxWorkerThreads; i++)
            {
                // 创建一个新线程,并把它添加到列表_workerThreads清单中。
                Thread newThread = new Thread(new ThreadStart(ProcessQueuedItems));
                _workerThreads.Add(newThread);

                // 对新线程的配置并启动它。
                newThread.Name = "ManagedPoolThread #" + i.ToString();
                newThread.IsBackground = true;
                newThread.Start();
            }
        }

        #endregion

        #region [ 公共函数 ]

        /// <summary>添加一个工作项到线程池列队中</summary>
        /// <param name="callback">一个线程池线程要执行的回调方法</param>
        public static void QueueUserWorkItem(WaitCallback callback)
        {
            QueueUserWorkItem(callback, null);
        }

        /// <summary>添加一个工作项到线程池列队中.</summary>
        /// <param name="callback">一个线程池线程要执行的回调方法</param>
        /// <param name="state">从线程池中送达时即传递给委托的对象。</param>
        public static void QueueUserWorkItem(WaitCallback callback, object state)
        {
            WaitingCallback waiting = new WaitingCallback(callback, state);
            lock (_poolLock)
            {
                _waitingCallbacks.Enqueue(waiting);
            }
            _workerThreadNeeded.AddOne();
        }

        /// <summary>清空线程池中的工作队列。重置池中的所有线程.</summary>
        public static void Reset()
        {
            lock (_poolLock)
            {
                // 清空所有等待的回调
                try
                {
                    // Try to dispose of all remaining state
                    foreach (object obj in _waitingCallbacks)
                    {
                        WaitingCallback callback = (WaitingCallback)obj;
                        if (callback.State is IDisposable)
                            ((IDisposable)callback.State).Dispose();
                    }
                }
                catch { }

                // 关闭所有现存的线程
                try
                {
                    foreach (Thread thread in _workerThreads)
                    {
                        if (thread != null) thread.Abort("reset");
                    }
                }
                catch { }

                // 重新初始化
                Initialize();
            }
        }

        #endregion

        #region [ 属性 ]

        /// <summary>获得在线程池中需要支配的最大线程数.</summary>
        public static int MaxThreads { get { return _maxWorkerThreads; } }

        /// <summary>获得当前活动的线程数目.</summary>
        public static int ActiveThreads { get { return _inUseThreads; } }

        /// <summary>获得所有等待去执行回调的列队的个数.</summary>
        public static int WaitingCallbacks { get { lock (_poolLock) { return _waitingCallbacks.Count; } } }

        #endregion

        #region [ 处理线程 ]

#if NET1
        /// <summary>Event raised when there is an exception on a threadpool thread.</summary>
        public static event UnhandledExceptionEventHandler UnhandledException;
#endif

        /// <summary>一个线程的辅助功能，处理从工作队列中的项目.</summary>
        private static void ProcessQueuedItems()
        {
            // 无限期地处理
            while (true)
            {
                //判断有无资源操作，如果有则执行【即从队列中取】，如没有则将当前线程置为等待。注意当前信息量中count数量与_waitingCallbacks队列长度相同。
                //通过QueueUserWorkItem的_workerThreadNeeded.AddOne();代码行来告之在当前线程池中处于wait的线程列表中找出一个线程来执行_waitingCallbacks列队中的任务（Dequeue）.
                //如果进程池中所有线程均为执行状态时，则系统还运行QueueUserWorkItem（）函数则可视其仅将任务放入队列即可，因为此处的while(true)会不断从队列中获取任务执行。
                _workerThreadNeeded.WaitOne();

                // 获取队列中的下一个项目。如果没有就sleep一会儿，直到我们唤醒正在等待回调。
                WaitingCallback callback = null;

                // 试图获得下一个可用的回调.  我们需要锁定队列,以便我们计数检查和检索原子.
                lock (_poolLock)
                {
                    if (_waitingCallbacks.Count > 0)
                    {
                        try
                        {
                            callback = (WaitingCallback)_waitingCallbacks.Dequeue();
                        }
                        catch { }
                    }
                }

                if (callback != null)
                {
                    // 有一个回调函数，执行它。确保准确记录有多少回调正在执行。
                    try
                    {
                        Interlocked.Increment(ref _inUseThreads);
                        callback.Callback(callback.State);
                    }
#if NET1
                    catch(Exception exc)
    				{
                        
                        //暂时注释下面的代码
                        try
                        {
                            UnhandledExceptionEventHandler handler = UnhandledException;
                            if (handler != null) handler(typeof(ManagedThreadPool), new UnhandledExceptionEventArgs(exc, false));
                        }
                        catch{}
                    }
#else
                    catch (Exception ex)
                    {
                        Function.WriteErrorLog(ex);
                    }
#endif
                    finally
                    {
                        Interlocked.Decrement(ref _inUseThreads);
                    }
                }
            }
        }
        #endregion

        /// <summary>用于控制一个回调的委托并声明那个委托.</summary>
        private class WaitingCallback
        {
            #region [ 成员变量 ]

            /// <summary>回调委托的回调.</summary>
            private WaitCallback _callback;

            /// <summary>回调委托的状态.</summary>
            private object _state;

            #endregion

            #region Construction

            /// <summary>初始化回调函数的委托.</summary>
            /// <param name="callback">回调委托的回调函数.</param>
            /// <param name="state">状态</param>
            public WaitingCallback(WaitCallback callback, object state)
            {
                _callback = callback;
                _state = state;
            }

            #endregion

            #region Properties

            /// <summary>获取回调委托的回调函数</summary>
            public WaitCallback Callback { get { return _callback; } }

            /// <summary>获取调用回调委托的状态</summary>
            public object State { get { return _state; } }

            #endregion
        }
    }

    /// <summary>Semaphore是基于Monitor类信号来实施的</summary>
    public class Semaphore
    {
        #region [ 成员变量 ]

        /// <summary>为Semaphore分配的单位数目。</summary>
        private int _count;

        private object _semLock = new object();

        #endregion

        #region [ 构造函数 ]

        /// <summary> Initialize the semaphore as a binary semaphore.</summary>
        public Semaphore()
            : this(1)
        {
        }

        /// <summary> Initialize the semaphore as a counting semaphore.</summary>
        /// <param name="count">可暂时理解为当前队列中要执行的任务数Initial number of threads that can take out units from this semaphore.</param>
        /// <exception cref="ArgumentException">Throws if the count argument is less than 0.</exception>
        public Semaphore(int count)
        {
            if (count < 0) throw new ArgumentException("Semaphore must have a count of at least 0.", "count");
            _count = count;
        }

        #endregion

        #region [ 同步操作 ]

        /// <summary>往Semaphore添加一个单元.</summary>
        public void AddOne() { V(); }

        /// <summary>往Semaphore取出一个单元.().</summary>
        public void WaitOne() { P(); }

        private void P()
        {
            // Lock so we can work in peace.  This works because lock is actually
            // built around Monitor.
            lock (_semLock)
            {
                // Wait until a unit becomes available.  We need to wait
                // in a loop in case someone else wakes up before us.  This could
                // happen if the Monitor.Pulse statements were changed to Monitor.PulseAll
                // statements in order to introduce some randomness into the order
                // in which threads are woken.
                while (_count <= 0)
                    Monitor.Wait(_semLock, Timeout.Infinite); //释放对象上的锁并阻止当前线程，直到它重新获取该锁，如果指定的超时间隔已过，则线程进入就绪队列
                _count--;
            }
        }

        private void V()
        {
            // Lock so we can work in peace.  This works because lock is actually
            // built around Monitor.
            lock (_semLock)
            {
                // Release our hold on the unit of control.  Then tell everyone
                // waiting on this object that there is a unit available.
                _count++;
                //向一个或多个等待线程发送信号。该信号通知等待线程锁定对象的状态已更改，并且锁的所有者准备释放该锁。
                //等待线程被放置在对象的就绪队列中以便它可以最后接收对象锁。一旦线程拥有了锁，它就可以检查对象的新状态以查看是否达到所需状态。
                Monitor.Pulse(_semLock);
            }
        }

        /// <summary>以指定的计数来重置信号，谨慎应用。</summary>
        public void Reset(int count)
        {
            lock (_semLock)
            {
                _count = count;
            }
        }

        #endregion
    }

}
